package com.ruirui.flowdemo;

import javax.annotation.processing.Processor;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {
    /**
     * 中间处理接口
     * 它即使发布者 又是订阅者
     * 为啥继承submissionpublisher？
     */
    static class FlowProcessor extends SubmissionPublisher<String>
            implements Flow.Processor<String, String> {
        private Flow.Subscription subscription ;
        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            System.out.println("FlowProcessor 订阅完成");
            this.subscription = subscription;
            subscription.request(1);
        }

        @Override
        public void onNext(String item) {
            System.out.println("FlowProcessor 接收到数据：" + item);
            item += "被我污染了";
            submit(item);// 重新作为发布者发送
            subscription.request(1); //  请求下一条数据

        }

        @Override
        public void onError(Throwable throwable) {
            System.out.println("FlowProcessor 接收到错误：" + throwable.getMessage());

        }

        @Override
        public void onComplete() {
            System.out.println("FlowProcessor 接收完成");

        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 创建一个发布者
//        Flow.Publisher<String> publisher = new Flow.Publisher<String>() {
//            private Flow.Subscriber<? super String> subscription;
//            /**
//             * 订阅者
//             * @param subscriber
//             */
//            @Override
//            public void subscribe(Flow.Subscriber<? super String> subscriber) {
//                // 保存订阅者
//                this.subscription = subscriber;
//            }
//        };
        //发布者发布数据到队列（jvm 内建的缓冲区）
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        /**
         * 订阅者
         * 订阅者订阅发布者
         * 订阅者接收发布者的数据
         */
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;
            /**
             * 订阅开始
             * @param subscription
             */
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread()+ "订阅开始了"+ subscription);
                this.subscription = subscription;
                //请求一个元素
                this.subscription.request(1);
            }
            /**
             * 下一个元素到达时候调用
             * @param item
             */

            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread()+"接收到数据："+ item);
                subscription.request(1);
                if ("hello5".equals(item))
                    subscription.cancel(); //取消订阅
            }
            /**
             * 错误处理
             * @param throwable
             */

            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread()+"错误处理："+ throwable.getMessage());

            }
            /**
             * 完成处理
             */
            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread()+"完成处理");

            }
        };

        //绑定发布者和订阅者
        FlowProcessor flowProcessor = new FlowProcessor(); // 中间商
        publisher.subscribe(flowProcessor);//订阅者
        flowProcessor.subscribe(subscriber);
        for (int i = 0; i < 10; i++) {
            publisher.submit("hello" + i);
        }

        publisher.close();
        Thread.sleep(6000);

    }
}
